package io.github.junxworks.junx.stream.extractor;

import com.google.common.collect.Lists;
import io.github.junxworks.junx.core.exception.BaseRuntimeException;
import io.github.junxworks.junx.core.util.StringUtils;
import io.github.junxworks.junx.event.EventBus;
import io.github.junxworks.junx.event.EventContext;
import io.github.junxworks.junx.stream.constants.ComponentStatus;
import io.github.junxworks.junx.stream.constants.DefaultParams;
import io.github.junxworks.junx.stream.extractor.convert.Convertor;
import io.github.junxworks.junx.stream.pipeline.BaseComponent;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Getter
@Setter
@Slf4j
public abstract class AbstractExtractor extends BaseComponent implements DataExtractor {
    protected EventBus eventBus;

    protected Convertor<?> convertor;

    protected String ouputTopic;

    protected List<Object> dataBoat;

    protected int totalCount = 0;

    @Override
    public boolean isDone() {
        return this.getStatus().equals(ComponentStatus.STOPPED);
    }

    /**
     * @throws Exception
     */
    @Override
    public void initialize() throws Exception {
    }

    protected void beforeExtract(ExtractContext context) {
        convertor = context.getConvertor();
        ouputTopic = context.getOutputTopic();
        if (StringUtils.isNull(ouputTopic)) {
            throw new BaseRuntimeException(this.flowName + "数据输出主题不能为空");
        }
        dataBoat = Lists.newArrayListWithCapacity(context.getBatchSize());
        this.start();
    }

    protected void flush() {
        for (Object data : dataBoat) {
            EventContext ctx = new EventContext(ouputTopic);
            ctx.setData(DefaultParams.PARAM_DATA, data);
            try {
                eventBus.publish(ctx);
            } catch (Exception e) {
                throw new RuntimeException(this.flowName + "发送数据抽取事件失败", e);
            }
        }
        totalCount += dataBoat.size();
        dataBoat.clear();
        log.info(this.flowName + "数据抽取事件当前发送记录数：" + totalCount);
    }

    /**
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
    }
}
